package com.gxljc.bear.crawler.weixin;

import com.gxljc.commons.util.AdvFile;
import com.gxljc.commons.util.ILineParser;
import com.gxljc.commons.util.StringUtil;
import com.gxljc.bear.crawler.base.DolphinCrawlerConf;
import com.gxljc.bear.crawler.base.DolphinCrawlerConsts;
import com.gxljc.bear.crawler.base.DolphinFetchData;
import com.gxljc.bear.crawler.itaogao.ItaogaoConst;
import com.gxljc.bear.crawler.proxy.ProxyUtil;
import com.gxljc.bear.crawler.util.SparkUtil;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.log4j.Logger;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.jsoup.Jsoup;
import org.jsoup.nodes.Document;
import org.jsoup.nodes.Element;
import org.jsoup.select.Elements;

import java.io.Serializable;
import java.net.URLEncoder;
import java.util.*;

/**
 * 搜狗微信种子账号数据爬虫。
 *
 * @author tanghaitao
 * @since 2022-8-1
 */
public class SogouAccountCrawler implements Serializable {
    public static Logger LOG = Logger.getLogger(SogouAccountCrawler.class);
    private Boolean proxy = false;
    SogouPageSeedCrawler crawler = null;
    private Boolean updateAll = false;

    public SogouAccountCrawler(Boolean proxy, Boolean updateAll) {
        this.proxy = proxy;
        crawler = new SogouPageSeedCrawler(proxy, updateAll);
    }

    public void crawl() throws Exception {
        final long SLEEP_TIME = 1 * 60 * 60 * 1000l;
        while (true) {
            List<AccountTable> seeds = null;
            try{
                seeds = WeixinUtil.getAccountSeed();
            }catch(Exception er){
                er.printStackTrace();
            }
            if (CollectionUtils.isEmpty(seeds)) {
                LOG.error("page is empty, sleep " + SLEEP_TIME);
                Thread.sleep(SLEEP_TIME); //没有数据，暂停
                continue;
            }
            int size = seeds.size();
            LOG.info("weixin account seeds size = " + size);
            ProxyUtil.initQueue(proxy);
            try {
                crawl(seeds);
            } catch (Exception er) {
                er.printStackTrace();
                LOG.error("spark job error", er);
            }
            LOG.info("sleep " + SLEEP_TIME);
            Thread.sleep(SLEEP_TIME);
        }
    }

    public void crawlNoSpark(List<AccountTable> seeds) throws Exception {
        if (CollectionUtils.isEmpty(seeds)) return;
        for (AccountTable seed : seeds) {
            try {
                crawl(seed);
                Thread.sleep(2000);
            } catch (Exception er) {
                er.printStackTrace();
            }
        }
    }

    public void crawl(List<AccountTable> seeds) throws Exception {
        int coreMax = 10;
        JavaSparkContext jsc = SparkUtil.createCommonsSparkContext(
                "bear-搜狗微信账号种子-" + seeds.size(), coreMax,
                coreMax * 2, SogouAccountCrawler.class);
        JavaRDD<AccountTable> seedsRDD = jsc.parallelize(new ArrayList<AccountTable>(seeds));
        long count = seedsRDD.mapPartitions(
                new FlatMapFunction<Iterator<AccountTable>, Integer>() {
                    private static final long serialVersionUID = 1L;

                    @Override
                    public Iterable<Integer> call(Iterator<AccountTable> seeds)
                            throws Exception {
                        int successCnt = 0;
                        while (seeds.hasNext()) {
                            try {
                                AccountTable seed = seeds.next();
                                int ret = crawl(seed);
                                if (ret == 1)
                                    successCnt++;
                                if (successCnt % 100 == 0) {
                                    LOG.info("successCnt=" + successCnt);
                                }
                            } catch (Exception er) {
                                er.printStackTrace();
                            }
                        }
                        return Arrays.asList(successCnt);
                    }
                }).count();
        LOG.info("all count=" + count);
        jsc.stop();
    }

    //单条url入口
    public int crawl(AccountTable table) throws Exception {
        LOG.info("crawl = " + table.accountId);
        String id = table.accountId;
        String url = String.format(WeixinConst.SEARCH_URL, id);
        DolphinFetchData fetchData = DolphinFetchData.getInstance(DolphinCrawlerConsts.CrawlerChannelType.MOVIE.getName());
        byte[] htmlByte = fetchData.getWeixinHtml(url, WeixinConst.CRAWLER_FIRST_URL, proxy, WeixinConst.CRAWL_TIMEOUT, null);
        if (htmlByte == null) {
            LOG.error("page result is empty url = " + url);
            return DolphinCrawlerConsts.RESULT_NO;
        }
        String html = new String(htmlByte);
        if (StringUtils.isEmpty(html)) {
            LOG.error("page result is empty url = " + url);
            return DolphinCrawlerConsts.RESULT_NO;
        }
        int flag = extract(table, html);
        if (flag == DolphinCrawlerConsts.RESULT_NO) {
            return DolphinCrawlerConsts.RESULT_NO;
        }
        crawler.crawl(table);
        return DolphinCrawlerConsts.RESULT_YES;
    }

    private String genCrawlUrl(String url) throws Exception {
        Configuration conf = DolphinCrawlerConf.getInstance();
        String webHost = conf.get(DolphinCrawlerConsts.PARAM_DETECTOR_WEB_HOST);
        if (StringUtils.isEmpty(webHost)) return url;
        String newUrl = URLEncoder.encode(url, "utf-8");
        return String.format(webHost + ItaogaoConst.DETECTOR_WEB_PATH, DolphinCrawlerConsts.PARAM_DETECTOR_WEB_ID, DolphinCrawlerConsts.PARAM_DETECTOR_WEB_TYPE, newUrl);
    }

    private int extract(AccountTable table, String html) {
        String id = table.accountId;
        Document doc = Jsoup.parse(html);
        Elements bodyElement = doc.select(".news-box ul li");
        if (CollectionUtils.isEmpty(bodyElement)) return DolphinCrawlerConsts.RESULT_NO;
        for (Element element : bodyElement) {
            Elements idElement = element.select(".info label");
            if (CollectionUtils.isEmpty(idElement)) continue;
            String matcherId = idElement.text();
            if (id.equalsIgnoreCase(matcherId)) {
                Elements aElement = element.select(".tit a");
                if (CollectionUtils.isEmpty(aElement)) continue;
                String url = aElement.attr("href");
                table.setCrawlUrl(url);
                return DolphinCrawlerConsts.RESULT_YES;
            }
        }
        return DolphinCrawlerConsts.RESULT_NO;
    }

    /**
     * 抽取二维码
     *
     * @param element
     * @return
     */
    private String extractQRCode(Element element) {
        Elements qrImgElements = element.select(".pos-ico div img");
        if (CollectionUtils.isEmpty(qrImgElements)) return null;
        for (Element img : qrImgElements) {
            String dataType = img.attr("data-type");
            if (StringUtils.isEmpty(dataType)) continue;
            if (dataType.equals("qr")) {
                String src = img.attr("src");
                return src.replaceAll("&quot;", "\"").replaceAll("&amp;", "&").replace("\\\\/", "/").replace("&nbsp;", " ");
            }
        }
        return null;
    }


}
